package com;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.sql.*;
import java.util.ArrayList;
import java.util.List;

/**
 * @Description: TODO QQ1667847363
 * @author: xiao kun tai
 * @date:2022/2/8 11:10
 *
 *
 * TODO Flink 流处理  （通用，不需要创建实体类）
 *
 * TODO Mysql的数据导入Hive表中    自定义Source，自定义Sink，缺陷：每次向Hive表插入1条数据，出现一次MR任务
 */
public class StreamMysqlToHiveCommon {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        System.setProperty("HADOOP_USER_NAME","root");

        DataStream<List<String>> streamSource = env.addSource(new MyMysqlSource("select * from test.mysql_hive"));
        streamSource.print();


        // todo insert into test.mysql_hive(id,name,age,money,todate,ts) values(66,"66",66,66,"2022-02-08","2022-02-08 16:07:50.0")
        streamSource.addSink(new MyHiveSink("insert into test.mysql_hive(id,name,age,money,todate,ts) values(?,?,?,?,?,?)"));
//        streamSource.addSink(HiveUtil.getSink("insert into test.mysql_hive(?,?,?,?,?,?)"));

        env.execute();


    }

    /**
     * CREATE TABLE `mysql_hive`  (
     * `id` int(11) NULL DEFAULT NULL,
     * `name` varchar(255) ,
     * `age` int(11) NULL DEFAULT NULL,
     * `money` double NULL DEFAULT NULL,
     * `todate` date NULL DEFAULT NULL,
     * `ts` timestamp NULL DEFAULT NULL
     * ) ;
     */
    public static class MyMysqlSource extends RichSourceFunction<List<String>> {
        private String sql;
        private Connection conn = null;
        private PreparedStatement pstm = null;
        private ResultSet rs = null;

        public MyMysqlSource(String sql) {
            this.sql = sql;
        }


        @Override
        public void open(Configuration parameters) throws Exception {
            String driver = "com.mysql.jdbc.Driver";
            String url = "jdbc:mysql://localhost:3306/test?useSSL=true";
            String username = "root";
            String password = "A";
            Class.forName(driver);
            conn = DriverManager.getConnection(url, username, password);
            pstm = conn.prepareStatement(sql);
        }

        @Override
        public void run(SourceContext<List<String>> sourceContext) throws Exception {
            rs = pstm.executeQuery();
            int count = rs.getMetaData().getColumnCount();
            ArrayList<String> bean = new ArrayList<>();
            while (rs.next()){
                bean.clear();
                for (int i = 1; i <= count; i++) {
                    bean.add(rs.getString(i));
                }
                sourceContext.collect(bean);
            }

        }

        @Override
        public void cancel() {

        }

        @Override
        public void close() throws Exception {
            if(rs!=null){
                try {
                    rs.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
            if(pstm!=null){
                try {
                    pstm.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
            if (conn != null) {
                try {
                    conn.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        }
    }


    /**
     * CREATE TABLE `mysql_hive`  (
     * `id` int,
     * `name` string ,
     * `age` int,
     * `money` double,
     * `todate` date,
     * `ts` timestamp
     * ) ;
     */
    public static class MyHiveSink extends RichSinkFunction<List<String>> {



        private PreparedStatement pstm;
        private Connection conn;
        private String sql;

        public MyHiveSink(String sql) {
            this.sql = sql;
        }

        @Override
        public void open(Configuration parameters) throws Exception {
            conn = getConnection();
            pstm = conn.prepareStatement(sql);

        }

        @Override
        public void invoke(List<String> value, Context context) throws Exception {
            for (int i = 1; i <= value.size(); i++) {
                pstm.setString(i,value.get(i-1));
            }
            pstm.executeUpdate();

        }

        @Override
        public void close()  {
            if (pstm != null) {
                try {
                    pstm.close();
                } catch (SQLException e) {

                }
            }
            if (conn != null) {
                try {
                    conn.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        }


        private static Connection getConnection() {
            Connection conn = null;
            try {
                String jdbc = "org.apache.hive.jdbc.HiveDriver";
//                String url = "jdbc:hive2://192.168.88.106:10000/test";
                String url = "jdbc:hive2://192.168.88.108:10000/test";
//                String url = "jdbc:hive2://122.7.87.134:20015/test";
                String user = "root";
                String password = "";
                Class.forName(jdbc);
                conn = DriverManager.getConnection(url, user, password);

            } catch (Exception e) {
                System.out.println(e.getMessage());
            }
            return conn;
        }
    }
}
